package com.coding4play.mqtt.broker.internal;

import com.coding4play.mqtt.common.entity.message.SubMessage;
import com.coding4play.mqtt.common.entity.session.ChannelSession;
import com.coding4play.mqtt.common.repository.IMessageIdService;
import com.coding4play.mqtt.common.repository.ISessionRepository;
import com.coding4play.mqtt.common.repository.ISubMessageRepository;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * 消息发送
 */
@Component
@AllArgsConstructor
@Slf4j
public class InternalBroadcast {

    private ISessionRepository sessionRepository;
    private ISubMessageRepository subMessageRepository;
    private IMessageIdService<Integer> messageIdService;

    /**
     * 发送消息到Topic(排除自身clientId)
     *
     * @param clientId
     * @param topic
     * @param mqttQoS
     * @param messageBytes
     * @param retain
     * @param dup
     */
    public void broadcastMessage(final String clientId, final String topic, final MqttQoS mqttQoS, byte[] messageBytes, boolean retain, boolean dup) {
        List<SubMessage> subClients = subMessageRepository.search(topic);

        subClients.stream()
                .filter(subClient -> !clientId.equals(subClient.getClientId()))
                .filter(subClient -> sessionRepository.hasSession(subClient.getClientId()))
                .peek(qos -> qos.setMqttQoS(Math.min(mqttQoS.value(), qos.getMqttQoS())))
                .forEach(subClient -> {
                    int respQos = subClient.getMqttQoS();
                    int messageId = MqttQoS.AT_MOST_ONCE.value() == respQos ? 0 : this.messageIdService.nextMessageId();
                    log.debug("PUBLISH - clientId: {}, topic: {}, Qos: {}, messageId: {}", subClient.getClientId(), topic, respQos, messageId);

                    MqttMessage mqttMessage = MqttMessageFactory.newMessage(
                            new MqttFixedHeader(MqttMessageType.PUBLISH, dup, MqttQoS.valueOf(respQos), retain, 0),
                            new MqttPublishVariableHeader(topic, messageId),
                            Unpooled.buffer().writeBytes(messageBytes));

                    ChannelSession channelSession = sessionRepository.get(subClient.getClientId());
                    Channel c0 = this.sessionRepository.getChannel(channelSession);
                    if (c0 != null) {
                        c0.writeAndFlush(mqttMessage);
                    }
                });


    }
}
